package cc.rengu.oltp.service.base;

import cc.rengu.oltp.service.common.constant.AppParamConstant;
import cc.rengu.oltp.service.common.constant.TreeNodeConstant;
import cc.rengu.oltp.service.common.dao.DstChannelInfoMapper;
import cc.rengu.oltp.service.common.dao.FusingCtrlInfoMapper;
import cc.rengu.oltp.service.common.dao.impl.DstChannelInfoMapperImpl;
import cc.rengu.oltp.service.common.dao.impl.FusingCtrlInfoMapperImpl;
import cc.rengu.oltp.service.common.entity.DstChannelInfo;
import cc.rengu.oltp.service.common.entity.FusingCtrlInfo;
import cc.rengu.oltp.service.common.enums.FusingCtrlTypeEnum;
import cc.rengu.oltp.service.common.enums.TransStatusEnum;
import cc.rengu.oltp.service.model.MonitorMode;
import cc.rengu.oltp.service.model.TransMonitorInfo;
import cc.rengu.oltp.service.realize.TransMonitorService;
import cc.rengu.oltp.service.realize.impl.TransMonitorServiceImpl;
import cc.rengu.oltp.utility.util.*;
import com.alibaba.fastjson.JSON;

import java.io.ByteArrayOutputStream;
import java.io.PrintWriter;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 交易熔断控制服务
 * Created by 王成 on 2020/2/18
 */
public class TransFusingService extends RadpService {
    @Override
    protected int bizInit(String s) {
        return 0;
    }

    @Override
    protected int bizProcess(String s, int i) {
        XmlTreeUtil xmlTreeUtil = new XmlTreeUtil();
        try {
            String instId = xmlTreeUtil.getXmlTreeStringValue(TreeNodeConstant.INST_ID);
            String transStatus = xmlTreeUtil.getXmlTreeStringValue(TreeNodeConstant.TRANS_STATUS);
            /* Step1：获取熔断控制信息 */
            FusingCtrlInfoMapper fusingCtrlInfoMapper = new FusingCtrlInfoMapperImpl();
            List<FusingCtrlInfo> fusingCtrlInfoList = fusingCtrlInfoMapper.selectAllFusingCtrlInfo(instId);
            if (null == fusingCtrlInfoList || fusingCtrlInfoList.isEmpty()) {
                rglog.info("法人行<{}>未配置熔断控制信息，不进行熔断控制!", instId);
                return 0;
            }
            /* Step2：交易失败时，判断是否触发熔断机制 */
            if (null != transStatus && transStatus.isEmpty() && !transStatus.equals(TransStatusEnum.TRANS_SUCC.getStatus())) {
                /* Step2.1：获取当前熔断状态为未熔断熔断控制记录 */
                fusingCtrlInfoList = fusingCtrlInfoList.stream().filter(item -> item.getFusingStatus().equals(AppParamConstant.NO)).collect(Collectors.toList());
                if (!fusingCtrlInfoList.isEmpty()) {
                    for (FusingCtrlInfo fusingCtrlInfo : fusingCtrlInfoList) {
                        fusingCtrlCheck(fusingCtrlInfo);
                    }
                }
            }
            /* Step3：熔断恢复巡检 */
            long currentTimeStamp = DateUtil.getCurrentTimeStamp();
            fusingCtrlInfoList = fusingCtrlInfoList.stream().filter(item -> item.getFusingStatus().equals(AppParamConstant.YES) &&
                    currentTimeStamp > DateUtil.getTimeStamp(item.getFusingEndTime(), "yyyy-MM-dd HH:mm:ss.SSS")).collect(Collectors.toList());
            if (!fusingCtrlInfoList.isEmpty()) {
                for (FusingCtrlInfo fusingCtrlInfo : fusingCtrlInfoList) {
                    fusingStatusRecovery(fusingCtrlInfo);
                }
            }
        } catch (Exception e) {
            ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
            e.printStackTrace(new PrintWriter(byteArrayOutputStream, true));
            rglog.error("交易监控及熔断控制服务处理失败，异常信息:<{}>", byteArrayOutputStream.toString());
            return -1;
        }
        return 0;
    }

    @Override
    protected int bizTerminate(String s) {
        return 0;
    }

    /**
     * 熔断状态恢复
     *
     * @param fusingCtrlInfo 熔断控制信息
     * @throws Exception 异常
     */
    private void fusingStatusRecovery(FusingCtrlInfo fusingCtrlInfo) throws Exception {
        /* Step1：根据熔断控制类型获取对应的交易缓存记录 */
        if (FusingCtrlTypeEnum.PAY_CHANNEL.getCtrlType().equals(fusingCtrlInfo.getFusingCtrlType())) {
            /* Step1.1：通道熔断控制 */
            DstChannelInfoMapper dstChannelInfoMapper = new DstChannelInfoMapperImpl();
            DstChannelInfo dstChannelInfo = dstChannelInfoMapper.selectDstChannelInfoByPrimaryKey(fusingCtrlInfo.getInstId(), fusingCtrlInfo.getFusingCtrlInst().trim());
            if (null != dstChannelInfo) {
                channelFusingStatusRecovery(dstChannelInfo, fusingCtrlInfo);
            }
        } else {
            rglog.error("暂不支持的熔断控制类型<{}>", fusingCtrlInfo.getFusingCtrlType());
        }
    }

    /**
     * 熔断控制检查
     *
     * @param fusingCtrlInfo 熔断控制信息
     * @throws Exception 异常
     */
    private void fusingCtrlCheck(FusingCtrlInfo fusingCtrlInfo) throws Exception {
        /* Step1：根据熔断控制类型获取对应的交易缓存记录 */
        if (FusingCtrlTypeEnum.PAY_CHANNEL.getCtrlType().equals(fusingCtrlInfo.getFusingCtrlType())) {
            /* Step1.1：通道熔断控制 */
            DstChannelInfoMapper dstChannelInfoMapper = new DstChannelInfoMapperImpl();
            DstChannelInfo dstChannelInfo = dstChannelInfoMapper.selectDstChannelInfoByPrimaryKey(fusingCtrlInfo.getInstId(), fusingCtrlInfo.getFusingCtrlInst().trim());
            if (null != dstChannelInfo) {
                channelFusingCtrlCheck(dstChannelInfo, fusingCtrlInfo);
            }
        } else {
            rglog.error("暂不支持的熔断控制类型<{}>", fusingCtrlInfo.getFusingCtrlType());
        }
    }

    /**
     * 支付通道熔断控制检查并更新
     *
     * @param dstChannelInfo 支付通道信息
     * @param fusingCtrlInfo 熔断控制信息
     * @throws Exception 异常
     */
    private void channelFusingCtrlCheck(DstChannelInfo dstChannelInfo, FusingCtrlInfo fusingCtrlInfo) throws Exception {
        XmlTreeUtil xmlTreeUtil = new XmlTreeUtil();
        /* Step1：获取交易监控信息 */
        MonitorMode monitorMode = new MonitorMode();
        monitorMode.setInstId(fusingCtrlInfo.getInstId());
        monitorMode.setMonitorTime(fusingCtrlInfo.getFusingInspectTime());
        monitorMode.setMsgDstId(fusingCtrlInfo.getFusingCtrlInst().trim());
        TransMonitorService transMonitorService = new TransMonitorServiceImpl();
        List<TransMonitorInfo> transMonitorInfoList = transMonitorService.subscribeTransMonitorInfo(monitorMode);
        if (null == transMonitorInfoList || transMonitorInfoList.isEmpty()) {
            rglog.info("{}秒内通道<{}>未发生交易或者未配置交易监控!", fusingCtrlInfo.getFusingInspectTime(), fusingCtrlInfo.getFusingCtrlInst().trim());
            return;
        }
        /* Step2：获取对应通道的交易总笔数及失败的交易笔数 */
        long totalTxnNum = transMonitorInfoList.stream().filter(item -> null != item.getChannelMonitorInfoList() && !item.getChannelMonitorInfoList().isEmpty() &&
                item.getChannelMonitorInfoList().stream().anyMatch(channelMonitorInfo -> channelMonitorInfo.getChannelId().trim().equals(fusingCtrlInfo.getFusingCtrlInst().trim()))).count();
        long failTxnNum = transMonitorInfoList.stream().filter(
                item -> !item.getTransStatus().equals(TransStatusEnum.TRANS_SUCC.getStatus()) && null != item.getChannelMonitorInfoList() && !item.getChannelMonitorInfoList().isEmpty() &&
                        item.getChannelMonitorInfoList().stream().anyMatch(channelMonitorInfo -> channelMonitorInfo.getChannelId().trim().equals(fusingCtrlInfo.getFusingCtrlInst().trim()) && !channelMonitorInfo.getTransStatus().equals(TransStatusEnum.TRANS_SUCC.getStatus()))).count();
        if (totalTxnNum < fusingCtrlInfo.getFusingTransLimit() || (failTxnNum / totalTxnNum) > (fusingCtrlInfo.getFusingFailRatio() / 100)) {
            rglog.info("{}秒内交易总笔数<{}>小于熔断控制最低交易笔数<{}>或者交易失败率<{}>低于设置的熔断比例<{}>", fusingCtrlInfo.getFusingInspectTime(),
                    totalTxnNum, fusingCtrlInfo.getFusingTransLimit(), failTxnNum / totalTxnNum, fusingCtrlInfo.getFusingFailRatio() / 100);
            return;
        }
        /* Step3：开启数据库事务 */
        String corporation = xmlTreeUtil.getXmlTreeStringValue(AppParamConstant.CORPORATION);
        String dbPoolName = XmlConfigUtil.getXmlConfigCacheKeyValue(AppParamConstant.DATABASE_CONF_NODE, AppParamConstant.OLTP_POOL_NAME);
        if (null != corporation && !corporation.isEmpty()) {
            String dbPoolCashNode = corporation.trim() + AppParamConstant.DATABASE_CONF_NODE;
            dbPoolName = XmlConfigUtil.getXmlConfigCacheKeyValue(dbPoolCashNode, AppParamConstant.OLTP_POOL_NAME);
        }
        DbsUtil dbsUtil = new DbsUtil(dbPoolName);
        dbsUtil.dbsBeginTransaction();
        /* Step4：更新熔断控制状态为熔断 */
        fusingCtrlInfo.setFusingStatus(AppParamConstant.YES);
        long currentTimeStamp = DateUtil.getCurrentTimeStamp();
        fusingCtrlInfo.setFusingBeginTime(DateUtil.getDateByTimeStamp(currentTimeStamp, "yyyy-MM-dd HH:mm:ss.SSS"));
        fusingCtrlInfo.setFusingEndTime(DateUtil.getDateByTimeStamp(currentTimeStamp + fusingCtrlInfo.getFusingInspectTime(), "yyyy-MM-dd HH:mm:ss.SSS"));
        FusingCtrlInfoMapper fusingCtrlInfoMapper = new FusingCtrlInfoMapperImpl();
        int iReturnCode = fusingCtrlInfoMapper.updateFusingCtrlInfoByPrimaryKey(fusingCtrlInfo);
        if (0 != iReturnCode) {
            rglog.error("更新通道<{}>的熔断状态为熔断失败,iReturnCode:<{}>", fusingCtrlInfo.getFusingCtrlInst(), iReturnCode);
            dbsUtil.dbsEndTransaction(false);
        }
        /* Step5：更新支付通道状态为停用 */
        dstChannelInfo.setDstChannelStatus(AppParamConstant.NO);
        DstChannelInfoMapper dstChannelInfoMapper = new DstChannelInfoMapperImpl();
        iReturnCode = dstChannelInfoMapper.updateDstChannelInfoByPrimaryKey(dstChannelInfo);
        if (0 != iReturnCode) {
            rglog.error("更新通道<{}>状态为停用失败,iReturnCode:<{}>", dstChannelInfo.getDstChannelId(), iReturnCode);
            dbsUtil.dbsEndTransaction(false);
        }
        /* Step6：更新Redis缓存 */
        String channelKey = dstChannelInfo.getInstId().trim() + dstChannelInfo.getDstChannelId().trim();
        RedisUtil.hashSet(AppParamConstant.CHANNEL_ISS_INFO_CACHE, channelKey, JSON.toJSONString(dstChannelInfo));
        /* Step7：结束数据库事务 */
        dbsUtil.dbsEndTransaction(true);
    }

    /**
     * 支付通道熔断状态恢复
     *
     * @param dstChannelInfo 支付通道信息
     * @param fusingCtrlInfo 熔断控制信息
     * @throws Exception 异常
     */
    private void channelFusingStatusRecovery(DstChannelInfo dstChannelInfo, FusingCtrlInfo fusingCtrlInfo) throws Exception {
        XmlTreeUtil xmlTreeUtil = new XmlTreeUtil();
        /* Step1：开启数据库事务 */
        String corporation = xmlTreeUtil.getXmlTreeStringValue(AppParamConstant.CORPORATION);
        String dbPoolName = XmlConfigUtil.getXmlConfigCacheKeyValue(AppParamConstant.DATABASE_CONF_NODE, AppParamConstant.OLTP_POOL_NAME);
        if (null != corporation && !corporation.isEmpty()) {
            String dbPoolCashNode = corporation.trim() + AppParamConstant.DATABASE_CONF_NODE;
            dbPoolName = XmlConfigUtil.getXmlConfigCacheKeyValue(dbPoolCashNode, AppParamConstant.OLTP_POOL_NAME);
        }
        DbsUtil dbsUtil = new DbsUtil(dbPoolName);
        dbsUtil.dbsBeginTransaction();
        /* Step2：更新熔断控制状态为非熔断 */
        fusingCtrlInfo.setFusingStatus(AppParamConstant.NO);
        FusingCtrlInfoMapper fusingCtrlInfoMapper = new FusingCtrlInfoMapperImpl();
        int iReturnCode = fusingCtrlInfoMapper.updateFusingCtrlInfoByPrimaryKey(fusingCtrlInfo);
        if (0 != iReturnCode) {
            rglog.error("更新通道<{}>的熔断状态为非熔断失败,iReturnCode:<{}>", fusingCtrlInfo.getFusingCtrlInst(), iReturnCode);
            dbsUtil.dbsEndTransaction(false);
        }
        /* Step3：更新支付通道状态为启用 */
        dstChannelInfo.setDstChannelStatus(AppParamConstant.YES);
        DstChannelInfoMapper dstChannelInfoMapper = new DstChannelInfoMapperImpl();
        iReturnCode = dstChannelInfoMapper.updateDstChannelInfoByPrimaryKey(dstChannelInfo);
        if (0 != iReturnCode) {
            rglog.error("更新通道<{}>状态为启用失败,iReturnCode:<{}>", dstChannelInfo.getDstChannelId(), iReturnCode);
            dbsUtil.dbsEndTransaction(false);
        }
        /* Step4：更新Redis缓存 */
        String channelKey = dstChannelInfo.getInstId().trim() + dstChannelInfo.getDstChannelId().trim();
        RedisUtil.hashSet(AppParamConstant.CHANNEL_ISS_INFO_CACHE, channelKey, JSON.toJSONString(dstChannelInfo));
        /* Step5：结束数据库事务 */
        dbsUtil.dbsEndTransaction(true);
    }
}
